package com.atguigu.dataStreamApi.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.KafkaSourceBuilder;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.common.TopicPartition;

import java.util.HashMap;

public class Flink03_KafkaSource {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        HashMap<TopicPartition,Long> offsets = new HashMap<>();

        offsets.put(new TopicPartition("topicG",1),14838L);

        //对接数据源,首先构建kafka连接器对象
            //1.KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
            //1.        .build();

            //2.todo 传参
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("hadoop102:9092,hadoop103:9092") //传kafka集群节点
                .setGroupId("flink0620")//传消费者组
                .setTopics("topicG")//消费主题
                .setValueOnlyDeserializer(new SimpleStringSchema())//反序列化
                .setStartingOffsets(
                      //OffsetsInitializer.earliest() //最早的偏移位置，存在数据位置消费
                      //OffsetsInitializer.latest()//最新的偏移量，从新发数据处消费
                      //OffsetsInitializer.committedOffsets()//这段代码的功能是设置Kafka消费者的起始偏移量为已提交的偏移量。
                        OffsetsInitializer.offsets(offsets)//指定偏移量
                )//消费偏移量
                .build();

        DataStreamSource<String> ds
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");

        ds.print();

        env.setParallelism(1);


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
